package com.alisoft.nano.seda.impl;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;

import org.springframework.jmx.export.annotation.AnnotationMBeanExporter;
import org.springframework.jmx.export.annotation.ManagedNotification;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.jmx.export.naming.SelfNaming;
import org.springframework.jmx.export.notification.NotificationPublisher;
import org.springframework.jmx.export.notification.NotificationPublisherAware;
import org.springframework.jmx.support.JmxUtils;

import com.alisoft.nano.seda.Event;
import com.alisoft.nano.seda.EventMachine;
import com.alisoft.nano.seda.Stage;
import com.alisoft.nano.seda.StageNotFoundException;
import com.alisoft.nano.seda.StageRuntimeException;
import com.alisoft.nano.seda.TimeoutException;
import com.alisoft.nano.seda.notification.EventMachineNotification;
import com.alisoft.nano.seda.notification.EventMachineNotificationListener;
import com.alisoft.nano.seda.notification.StageNotificationListener;

/**
 * Spring implementation of EventMachine
 * 
 * @author zhimin.chenzm, anping.yinap
 *
 * @param <T>
 */
@SuppressWarnings("serial")
@ManagedResource(description = "Default Work Stage Manager")
@ManagedNotification(name = "Stage Notification", notificationTypes = {
		EventMachineNotification.ACCEPTED, EventMachineNotification.ERROR,
		EventMachineNotification.COMPLETED })
public class DefaultEventMachine<T> extends StageNotificationListener<T>
		implements EventMachine<T>, NotificationPublisherAware, SelfNaming {

    private String mBeanDomain = "com.alisoft.nano.seda";
    private String mBeanName = "event-machine";

    
    private List<EventMachineNotificationListener<T>> listeners = new ArrayList<EventMachineNotificationListener<T>>();
    
	private Map<String, Stage<T>> stages = new HashMap<String, Stage<T>>();
	
	private volatile boolean started = false;

	private NotificationPublisher publisher;

	public DefaultEventMachine() {
	}

	@ManagedOperation
	public synchronized void start() {
		if (this.started) {
			// already started
			return;
		}

		for (Stage<T> stage : stages.values()) {
			stage.start();
		}
		
		// register JMX 
        AnnotationMBeanExporter mBeanExporter = new AnnotationMBeanExporter();
        MBeanServer server = JmxUtils.locateMBeanServer();
        mBeanExporter.setServer(server);
        
        try {
            
            ObjectName machineName = this.getObjectName();
            mBeanExporter.registerManagedResource(this, machineName);
            
            for (EventMachineNotificationListener<T> listener : listeners) {
                server.addNotificationListener(machineName, listener, listener, null);
            }
            
        } catch (Exception e) {
            throw new StageRuntimeException("Failed to start the event machine", e);
        }
		
		this.started = true;
	}

	@ManagedOperation
	public synchronized void stop() {
		this.started = false;
		for (Stage<T> stage : stages.values()) {
			stage.stop();
		}
	}

	public boolean isStarted() {
		return this.started;
	}

	public void setStages(List<Stage<T>> stages) {
	    if (this.started) {
	        throw new IllegalStateException("The event machine is already started");
	    }
		for (Stage<T> stage : stages) {
		    if (this.stages.containsKey(stage.getName())) {
		        throw new IllegalArgumentException("The stage name " + stage.getName() + " is duplicated");
		    }
			this.stages.put(stage.getName(), stage);
			stage.addStageNotificationListener(this);
		}
	}
	
	
	public void addStage(Stage<T> stage) {
	    if (this.started) {
            throw new IllegalStateException("The event machine is already started");
        }
	    if (this.stages.containsKey(stage.getName())) {
            throw new IllegalArgumentException("The stage name " + stage.getName() + " is duplicated");
        }
	    this.stages.put(stage.getName(), stage);
	    stage.addStageNotificationListener(this);
	}
	
	public void addEventMachineNotificationListener(EventMachineNotificationListener<T> listener) {
	    this.listeners.add(listener);
	}
	
	public void setListeners(List<EventMachineNotificationListener<T>> listeners) {
        this.listeners = listeners;
    }
	
	
	public void setNotificationPublisher(NotificationPublisher publisher) {
		this.publisher = publisher;
	}

	/**
	 * The event is done by the previous stage.
	 */
	@Override
	protected void onDone(Event<T> event) {
		forwardEvent(event);
	}
	public String getMBeanName() {
        return mBeanName;
    }

    public void setMBeanName(String beanName) {
        mBeanName = beanName;
    }

    public String getMBeanDomain() {
        return mBeanDomain;
    }

    public void setMBeanDomain(String beanDomain) {
        mBeanDomain = beanDomain;
    }
    
    public ObjectName getObjectName() throws MalformedObjectNameException {
        return new ObjectName(this.mBeanDomain + ":name=" + this.mBeanName);
    }
	/**
	 * Forward this event to the next stage if necessary
	 * 
	 * @param event
	 */
	private void forwardEvent(Event<T> event) {
		if (event.getNextStage() == null || event.getException() != null) {
			// Event completed or exception occurred
		    finishEvent(event);
			return;
		}

		// get the next stage 
		Stage<T> stage = this.stages.get(event.getNextStage());
		if (stage == null) {
			// TODO log
			event
					.setException(new StageNotFoundException(event
							.getNextStage()));
			finishEvent(event);
			return;
		}

		try {
		    // post the event to the next stage
			boolean success = stage.post(event);
			if (!success) {
				event.setException(new TimeoutException(stage.getName(), stage
						.getDefaultTimeout(), TimeUnit.SECONDS));
				finishEvent(event);
			}
		} catch (InterruptedException e) {
			// Will never happen logically
			// TODO log & notify
			throw new IllegalStateException(e);
		}
	}
	
	private void finishEvent(Event<T> event) {
        if (event.getException() != null) {
            publisher.sendNotification(EventMachineNotification.newError(event));
        } else {
            publisher.sendNotification(EventMachineNotification.newCompleted(event));
        }
    }

	public void post(String stageName, T workUnit) throws InterruptedException {
		Event<T> event = new Event<T>(workUnit, stageName);		
		this.forwardEvent(event);
		this.publisher.sendNotification(EventMachineNotification
				.newAccepted(event));
	}    
}
